# rule_engine.py
from __future__ import annotations

import logging
import threading
from datetime import datetime, timedelta, time
from decimal import Decimal
from typing import Dict, List, NamedTuple, Optional, Set

from common.base_enums import MeterType
from common.database import DatabaseManager
from mqtt.consumer.mqtt_models import TelemetryRow

logger = logging.getLogger(__name__)

# ---------- 配置 ----------
# 1. 连续量映射  硬件原始键 -> 内部规则字段
RAWKEY_2_FIELD = {
	# 三相
	"a_v": "voltage", "b_v": "voltage", "c_v": "voltage",
	"a_c": "current", "b_c": "current", "c_c": "current",
	"a_p": "watt", "b_p": "watt", "c_p": "watt",
	"total_energy": "total_kwh",
	"T1": "temperature1", "T2": "temperature2", "T3": "temperature2",  # t3 复用
	# 单相
	"V": "voltage", "C": "current", "W": "watt",
}

FIELD_SEMANTIC_UNITS = {
	"voltage": {"V"},
	"current": {"A"},
	"watt": {"W", "kW", "VA", "var", "kvar"},
	"temperature1": {"℃"},
	"temperature2": {"℃"},
}

# 2. 离散事件键（规则里 unit=hr/HR/comm_lost 等）
EVENT_KEYS = {"hr", "HR", "comm_lost", "cover_open"}

REFRESH_INTERVAL = timedelta(seconds=30)


class Hit(NamedTuple):
	rule_code: str
	severity: int


class RuleEngine:
	"""单例，线程安全；依赖注入 DatabaseManager"""
	_instance: Optional["RuleEngine"] = None
	_lock = threading.Lock()

	def __new__(cls, db: Optional[DatabaseManager] = None, *args, **kwargs):
		# 双重检查锁定
		if cls._instance is None:
			with cls._lock:
				if cls._instance is None:
					if db is None:
						raise RuntimeError("RuleEngine 须通过 DatabaseManager 实例化")
					cls._instance = super().__new__(cls)
					cls._instance.db = db
					cls._instance._last_refresh = datetime.min
					cls._instance._rules: List[dict] = []
					cls._instance._index: Dict[str, List[dict]] = {}
		return cls._instance

	# ---------------- 对外 API ----------------
	def evaluate(self, raw_key: str, value, msg_id: str, meter_type: str) -> Optional[Hit]:
		try:
			self._refresh_if_needed()
		except Exception as e:
			# 任何异常都不影响后续逻辑
			logger.warning("刷新异常，使用缓存规则: %s", e)

		field = RAWKEY_2_FIELD.get(raw_key)
		if field:
			for unit in self._field_to_units.get(field, []):
				for r in self._index.get(unit, []):
					# 检查规则是否适用于当前电表类型
					if r["meter_type"] != "ALL" and r["meter_type"] != meter_type:
						continue
					if r["rule_type"] == "THRESHOLD" and self._match(Decimal(str(value)), r):
						if self._exists_alarm(msg_id, r["rule_code"]):
							return None
						return Hit(r["rule_code"], int(r["severity"]))

		if raw_key in EVENT_KEYS:
			for r in self._index.get(raw_key, []):
				if r['rule_type'] == 'EVENT' and str(value) == r['event_value']:
					if self._exists_alarm(msg_id, r['rule_code']):
						return None
					return Hit(r['rule_code'], int(r['severity']))
		return None

	# ---------------- 内部辅助 ----------------
	def _exists_alarm(self, msg_id: str, rule_code: str) -> bool:
		sql = """
	        SELECT 1
	        FROM am_alarm
	        WHERE msg_id = %s
	          AND alarm_type = %s
	        LIMIT 1
	    """
		rows = self.db.execute_query(sql, (msg_id, rule_code), commit=False)
		return bool(rows)

	def _refresh_if_needed(self) -> None:
		if datetime.now() - self._last_refresh < REFRESH_INTERVAL:
			return
		# 最多 3 次重试
		for attempt in range(3):
			try:
				rules = self._load_rules_safe()
				self._rules = rules
				self._index = self._build_index(rules)
				self._last_refresh = datetime.now()
				logger.info("RuleEngine 刷新成功，规则条数=%d", len(rules))
				return
			except Exception as e:
				logger.warning("刷新规则失败 %d/3: %s", attempt + 1, e)
				if attempt == 2:
					logger.error("已达最大重试，继续使用上次缓存")
				else:
					time.sleep(0.5 * (attempt + 1))
		# 如果仍失败，至少保证 _rules/_index 不为空（用缓存）
		if not self._rules:
			self._rules = []
			self._index = {}

	def _load_rules_safe(self) -> List[dict]:
		try:
			sql = """
			            SELECT rule_code, severity, threshold, unit, operator, rule_type, event_value, meter_type
			            FROM am_alarm_rule
			            WHERE enabled = 1
			            ORDER BY severity DESC
			        """
			return self.db.execute_query(sql, commit=False)
		except Exception as e:
			logger.error("加载规则失败: %s", e)
			return []

	def _build_index(self, rules: List[dict]) -> Dict[str, List[dict]]:
		idx: Dict[str, List[dict]] = {}
		self._field_to_units: Dict[str, Set[str]] = {}
		for r in rules:
			unit = r["unit"]
			idx.setdefault(unit, []).append(r)

			# 连续量规则：建立 field → unit 映射
			if r["rule_type"] == "THRESHOLD":
				unit = r["unit"]
				for raw_key, field in RAWKEY_2_FIELD.items():
					if field in FIELD_SEMANTIC_UNITS and unit in FIELD_SEMANTIC_UNITS[field]:
						self._field_to_units.setdefault(field, set()).add(unit)
		return idx

	@staticmethod
	def _match(val: Decimal, rule: dict) -> bool:
		op, thd = rule["operator"], Decimal(str(rule["threshold"]))
		return (
				(op == ">" and val > thd) or
				(op == "<" and val < thd) or
				(op == ">=" and val >= thd) or
				(op == "<=" and val <= thd) or
				(op == "==" and val == thd)
		)

	def any_hit(self, row: TelemetryRow) -> bool:
		meter_type = MeterType.SINGLE.value if row.meter_type == MeterType.SINGLE else MeterType.THREE_PHASE.value
		# 1. 连续量
		if self.evaluate_row_fields(row, row.msg_id, meter_type):
			return True
		# 2. 离散事件
		if self.evaluate_snapshot_events(row.raw_snapshot, row.msg_id, meter_type):
			return True
		return False

	def evaluate_row_fields(self, row_like, msg_id: str, meter_type: str) -> List[Hit]:
		"""对连续量字段统一做 evaluate，返回所有命中规则"""
		hits = []
		reverse_map = {v: k for k, v in RAWKEY_2_FIELD.items() if k not in EVENT_KEYS}
		for field in ("voltage", "current", "watt", "temperature1", "temperature2"):
			val = getattr(row_like, field, None)
			if val is None:
				continue
			raw_key = reverse_map.get(field)
			if not raw_key:
				continue
			hit = self.evaluate(raw_key, val, msg_id, meter_type)
			if hit:
				hits.append(hit)
		return hits

	def evaluate_snapshot_events(self, snapshot: dict, msg_id: str, meter_type: str) -> List[Hit]:
		"""对快照中所有离散事件键做 evaluate，返回所有命中规则"""
		hits = []
		for key in snapshot:
			if key in EVENT_KEYS:
				hit = self.evaluate(key, snapshot[key], msg_id, meter_type)
				if hit:
					hits.append(hit)
		return hits
